package com.sh.data.engine.domain.dataasset.job;

import com.sh.data.engine.domain.dataasset.model.domain.TblPropEntityKey;
import com.sh.data.engine.domain.dataasset.service.TblPropService;
import com.sh.data.engine.domain.shims.db.BaseDbManager;
import com.sh.data.engine.domain.shims.db.model.TableInfoDomain;
import com.sh.data.engine.domain.workspace.index.model.domain.ProjectDomain;
import com.sh.data.engine.domain.workspace.index.service.ProjectService;
import com.sh.data.engine.domain.workspace.manager.service.impl.StorageServiceImpl;
import com.sh.data.engine.job.core.context.XxlJobHelper;
import com.sh.data.engine.job.core.handler.annotation.XxlJob;
import com.sh.data.engine.repository.dao.dataasset.entity.TblPropEntity;
import com.sh.data.engine.repository.dao.dataasset.entity.TblPropWrapper;
import com.sh.data.engine.repository.dao.workspace.manager.entity.StorageEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.sql.SQLException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Author: zigui.zdf @Date: 2022/1/6 22:57:13 @Description: 同步存储层的表到虎符 ,每5分钟执行一次
 */
@Component
@Slf4j
public class SyncTblInfoToEngineTask {

    @Autowired
    private ProjectService projectService;

    @Autowired
    private StorageServiceImpl storageServiceImpl;

    @Autowired
    private TblPropService tblPropService;

    @XxlJob("SyncTblInfoToHufuTask")
    public void run() {
        final List<ProjectDomain> projectDomains = projectService.getAllProject();

        if (CollectionUtils.isEmpty(projectDomains)) {
            XxlJobHelper.log("当前工作空间数量为0，本次同步结束...");
            return;
        }

        XxlJobHelper.log("当前工作空间数量为：{}", projectDomains.size());

        for (ProjectDomain projectDomain : projectDomains) {

            XxlJobHelper.log("正在同步：[{}]空间内的表", projectDomain.getName());

            long s = System.currentTimeMillis();

            Long projectId = projectDomain.getId();
            BaseDbManager dbManager = null;
            try {
                dbManager = projectService.getDbManager(projectId, 10);
                run1(projectId, dbManager);

                XxlJobHelper.log(
                    "[{}]空间内的表同步结束，耗时:[{}]ms", projectDomain.getName(), System.currentTimeMillis() - s);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                XxlJobHelper.log(
                    "[{}]空间内的表同步异常，耗时:[{}]ms，原因：{}",
                    projectDomain.getName(),
                    System.currentTimeMillis() - s,
                    ExceptionUtils.getStackTrace(e));
            } finally {
                //  hiveManager需要关闭hiveClient
                if (null != dbManager) {
                    dbManager.close();
                }
            }
        }
    }

    /**
     * 同步逻辑： 1.查询出数据库中所有的表信息 2.查询出项目内所有的表信息tbl_prop 3.以第1步的数据为基准，
     *
     * @param projectId
     */
    private void run1(Long projectId, BaseDbManager dbManager) throws SQLException {

        final StorageEntity storageEntity = storageServiceImpl.getStorageInfoByProjectId(projectId);

        if (null == dbManager || null == storageEntity) {
            XxlJobHelper.log("空间未配置存储，同步结束...");
            return;
        }

        List<TblPropEntity> tblPropEntities = tblPropService.getAllTblPropByProjectId(projectId);

        if (null == tblPropEntities) {
            tblPropEntities = new ArrayList<>();
        }

        XxlJobHelper.log("hufu记录的空间内存储中共有:{}张表和视图", tblPropEntities.size());

        List<TblPropEntity> insertList = new ArrayList<>();
        List<TblPropEntity> updateList = new ArrayList<>();

        List<TableInfoDomain> tableAndViewList =
            dbManager.getTableAndViewList(storageEntity.getRmdbs().getDbName());

        if (null == tableAndViewList) {
            XxlJobHelper.log("空间内存储表和视图获取失败");
            tableAndViewList = new ArrayList<>();
        }
        XxlJobHelper.log("空间内存储中共有:{}张表和视图", tableAndViewList.size());

        Set<TblPropWrapper> uniqueObjects = new HashSet<>();
        for (TblPropEntity tblPropEntity : tblPropEntities) {

            TblPropWrapper tblPropWrapper = new TblPropWrapper(tblPropEntity.getId(),
                tblPropEntity.getProjectId(), tblPropEntity.getSchema(), tblPropEntity.getTableName(),
                tblPropEntity.getRowState(), tblPropEntity.getDataSecretId());
            if (uniqueObjects.contains(tblPropWrapper) && Objects.nonNull(
                tblPropWrapper.getDataSecretId())) {
                uniqueObjects.remove(tblPropWrapper);
                uniqueObjects.add(tblPropWrapper);
            } else if (!uniqueObjects.contains(tblPropWrapper)) {
                uniqueObjects.add(tblPropWrapper);
            }
        }
        List<Long> tblPropIds = tblPropEntities.stream().map(e -> e.getId())
            .collect(Collectors.toList());
        List<Long> keepListIds = uniqueObjects.stream().map(e -> e.getId())
            .collect(Collectors.toList());
        List<Long> longs = CollectionUtils.removeAll(tblPropIds, keepListIds).stream().collect(
            Collectors.toList());
        tblPropService.removeBatchByIds(longs);

        List<TblPropEntity> tblPropEntitiesNew = tblPropService.getAllTblPropByProjectId(projectId);

        final Map<TblPropEntityKey, TblPropEntity> tblPropEntityMap =
            tblPropEntitiesNew.stream()
                .collect(
                    Collectors.toMap(
                        x -> {
                            TblPropEntityKey tblPropEntityKey = new TblPropEntityKey();
                            tblPropEntityKey.setSchema(
                                StringUtils.isEmpty(x.getSchema()) ? "" : x.getSchema());
                            tblPropEntityKey.setTableName(x.getTableName());
                            return tblPropEntityKey;
                        },
                        x -> x));

        final Map<TblPropEntityKey, TableInfoDomain> tableInfoDomainMap =
            tableAndViewList.stream()
                .collect(
                    Collectors.toMap(
                        x -> {
                            TblPropEntityKey tblPropEntityKey = new TblPropEntityKey();
                            tblPropEntityKey.setSchema(
                                StringUtils.isEmpty(x.getSchema()) ? "" : x.getSchema());
                            tblPropEntityKey.setTableName(x.getTableName());
                            return tblPropEntityKey;
                        },
                        x -> x));

        for (TableInfoDomain tableInfoDomain : tableAndViewList) {
            final String tableName = tableInfoDomain.getTableName();
            TblPropEntityKey tblPropEntityKey = new TblPropEntityKey();
            tblPropEntityKey.setSchema(
                StringUtils.isEmpty(tableInfoDomain.getSchema()) ? "" : tableInfoDomain.getSchema());
            tblPropEntityKey.setTableName(tableName);
            // 遍历存储中的表, 判断是否已经被hufu同步过, 未同步过放入待插入表
            if (!tblPropEntityMap.containsKey(tblPropEntityKey)) {
                TblPropEntity tblPropEntity = new TblPropEntity();
                tblPropEntity.setProjectId(projectId);
                tblPropEntity.setSchema(tableInfoDomain.getSchema());
                tblPropEntity.setTblId(
                    tableInfoDomain.getTblId() == null
                        ? tblPropService.generateTblId(projectId)
                        : tableInfoDomain.getTblId());
                tblPropEntity.setTableName(tableName);
                tblPropEntity.setLifecycle(-1);
                tblPropEntity.setTblType(
                    tableInfoDomain.getTblType() == null ? "" : tableInfoDomain.getTblType().name());
                tblPropEntity.setRowState(1);
                insertList.add(tblPropEntity);
            } else {
                // 已经同步到hufu中, 判断是否需要更新
                final TblPropEntity tblPropEntity = tblPropEntityMap.get(tblPropEntityKey);
                boolean needUpdate = false;
                if (tblPropEntity.getRowState() == 0) {
                    tblPropEntity.setRowState(1);
                    needUpdate = true;
                }
                if (null == tblPropEntity.getTblId()) {
                    tblPropEntity.setTblId(
                        tableInfoDomain.getTblId() == null
                            ? tblPropService.generateTblId(projectId)
                            : tableInfoDomain.getTblId());
                    needUpdate = true;
                }
                // update comment
                if (!Objects.equals(tblPropEntity.getTableComment(), tableInfoDomain.getTableComment())) {
                    tblPropEntity.setTableComment(tableInfoDomain.getTableComment());
                    needUpdate = true;
                }
                if (needUpdate) {
                    updateList.add(tblPropEntity);
                }
            }
        }

        for (TblPropEntity tblPropEntity : tblPropEntities) {
            final String tableName = tblPropEntity.getTableName();
            final String schema = Optional.ofNullable(tblPropEntity.getSchema()).orElse("");
            TblPropEntityKey tblPropEntityKey = new TblPropEntityKey(schema, tableName);
            if (!tableInfoDomainMap.containsKey(tblPropEntityKey) && tblPropEntity.getRowState() == 1) {
                XxlJobHelper.log("delete {}", tblPropEntity.getId());
                tblPropEntity.setRowState(0);
                updateList.add(tblPropEntity);
            }
        }

        XxlJobHelper.log("本次同步，需新增{}条数据，需更新{}条数据", insertList.size(), updateList.size());

        if (!insertList.isEmpty()) {
            tblPropService.insertBatch(insertList);
        }

        if (!updateList.isEmpty()) {
            tblPropService.updateBatch(updateList);
        }
    }
}
